package spark.core.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object rddPersistence {
  // Todo RDD 持久化

  /** RDD通过Cache或者Persist方法将前面的计算结果缓存，
   * 默认情况下会把数据以缓存在JVM的堆内存中。但是并不是这两个方法被调用时立即缓存，而是触发后面的action算子时，
   * 该RDD将会被缓存在计算节点的内存中，并供后面重用。
   * */
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rddPersistence")
    val sc: SparkContext = new SparkContext(sparkConf)
    // Todo 1. RDD Cache 缓存

    val fileRDD: RDD[(String, Int)] = sc.textFile("Input/test.txt").flatMap(_.split(" ")).map((_, 1))

    // 数据缓存
    fileRDD.cache()

    // 可以更改存储级别
    // mapRdd.persist(StorageLevel.MEMORY_AND_DISK_2)
    /** object StorageLevel {
     * val NONE = new StorageLevel(false, false, false, false)
     * val DISK_ONLY = new StorageLevel(true, false, false, false)
     * val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
     * val MEMORY_ONLY = new StorageLevel(false, true, false, true)
     * val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
     * val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
     * val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
     * val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
     * val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
     * val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
     * val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
     * val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
     * */
    val resultRDD: RDD[(String, Int)] = fileRDD.reduceByKey(_ + _)
    // cache 操作会增加血缘关系，不改变原有的血缘关系
    println(resultRDD.toDebugString)

    /**
     * 缓存有可能丢失，或者存储于内存的数据由于内存不足而被删除，RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。
     * 通过基于RDD的一系列转换，丢失的数据会被重算，由于RDD的各个Partition是相对独立的，
     * 因此只需要计算丢失的部分即可，并不需要重算全部Partition。
     * */

    /**   Spark 会自动对一些Shuffle操作的中间数据做持久化操作(比如：reduceByKey)。这样
     *  做的目的是为了当一个节点Shuffle失败了避免重新计算整个输入。但是，在实际使用的时
     *  候，如果想重用数据，仍然建议调用persist或cache。
     *  */
  }
}
